package com.shujia.flink.sql;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class Demo1WordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建flink sql流处理环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        /*
         * 1、创建一个流
         */
        DataStream<String> words = env.socketTextStream("master", 8888);


        /*
         * 2、在流上定义表
         */
        tEnv.createTemporaryView("words", words);

        /*
         * 3、基于动态表执行连续查询
         */

        Table table = tEnv.sqlQuery("select f0 as word , count(1) as c from words group by f0");

        table.printSchema();

        /*
         * 4、将结果表转换成流
         */
        DataStream<Row> dataStream = tEnv.toChangelogStream(table);

        dataStream.print();

        env.execute();


    }
}
